package com.hopu.user.sex;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.sex.Sex;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.IOException;
import java.util.List;

public class SexAna {
    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setAppName("").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        JavaPairRDD<Sex, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String sex = json.getString("sex");
            String type = json.getString("event_type");

            return new Tuple2<>(new Sex(sex, type), 1);
        });

//        pairRDD.foreach(t -> System.out.println(t));

        //过滤不同性别的RDD
        JavaPairRDD<Sex, Integer> maleRDD = pairRDD.filter(t -> "男".equals(t._1.getSex()));
        JavaPairRDD<Sex, Integer> femaleRDD = pairRDD.filter(t -> "女".equals(t._1.getSex()));

        Sex maleCount = getRDDByType(maleRDD);
        System.out.println(maleCount);
        Sex femaleCount = getRDDByType(femaleRDD);
        System.out.println(femaleCount);


        //写入到HBase中
        String[] columns = {"sex", "view", "cart", "parchase", "total"};
        String[] values1 = {maleCount.getSex(), maleCount.getView()+"", maleCount.getCart()+"",
                maleCount.getParchase()+"", maleCount.getTotal()+""};
        String[] values2 = {femaleCount.getSex(), femaleCount.getView()+"", femaleCount.getCart()+"",
                femaleCount.getParchase()+"", femaleCount.getTotal()+""};

        try {
            HBaseUtils.putOneRowToHbase("totalana", maleCount.getSex(), "sexinfo", columns, values1);
            HBaseUtils.putOneRowToHbase("totalana", femaleCount.getSex(), "sexinfo", columns, values2);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public static Sex getRDDByType(JavaPairRDD<Sex, Integer> rdd) {
        //获取某个性别的不同操作的RDD
        JavaPairRDD<Sex, Integer> viewRDD = rdd.filter(t -> "view".equals(t._1.getType()));
        JavaPairRDD<Sex, Integer> cartRDD = rdd.filter(t -> "cart".equals(t._1.getType()));
        JavaPairRDD<Sex, Integer> parchasRDD = rdd.filter(t -> "parchase".equals(t._1.getType()));

        Sex viewSex = getGroupByType(viewRDD);
//        System.out.println(viewSex);
        Sex cartSex = getGroupByType(cartRDD);
        Sex parchaseSex = getGroupByType(parchasRDD);

        viewSex.setCart(cartSex.getCart());
        viewSex.setParchase(parchaseSex.getParchase());
        viewSex.setTotal(viewSex.getView()+cartSex.getCart()+parchaseSex.getParchase());

        return viewSex;
    }

    public static Sex getGroupByType(JavaPairRDD<Sex, Integer> rdd) {
        //分组统计不同操作类型下RDD的数量
        JavaPairRDD<Sex, Iterable<Integer>> viewGroup = rdd.groupByKey();
        JavaRDD<Sex> map = viewGroup.map(t -> {
            int count = 0;
            for (int v : t._2) {
                count += v;
            }
            if (t._1.getType().equals("view")) {
                t._1.setView(count);
            } else if (t._1.getType().equals("cart")) {
                t._1.setCart(count);
            } else {
                t._1.setParchase(count);
            }
            return t._1;
        });
        List<Sex> take = map.take(1);
        return take.get(0);
    }

}
